package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022,
                4,
                "dws_traffic_vc_ch_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    /**
     *
     * @param env
     * @param kafkaStrDS dwd_traffic_page
     *  {   "common":{"sc":"1","ar":"10","uid":"1168","os":"iOS 13.2.9","ch":"Appstore","is_new":"0","md":"iPhone X","mid":"mid_376","vc":"v2.1.134","ba":"iPhone","sid":"b0fad428-a8b8-452d-8a14-6344e9e1aff1"},
     *      "page":{"page_id":"course_list","item":"java","during_time":14470,"item_type":"keyword","last_page_id":"home"},
     *      "ts":1731500497559
     *                   }
     *
     */

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS
                = kafkaStrDS.map(JSON::parseObject);
        // jsonObjDS.print();

        //TODO 2.按照设备的id进行分组
        KeyedStream<JSONObject, String> midKeyedDS
                = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        // midKeyedDS.print();

        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = midKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor
                                = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
                        //设置状态的保留时间
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        //获取当前设备上次访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        //获取当前设备这次访问日期
                        Long ts = jsonObj.getLong("ts");
                        String curVisitDate = DateFormatUtil.tsToDate(ts);
                        Long uvCt = 0L;

                        if (StringUtils.isEmpty(lastVisitDate) || !curVisitDate.equals(lastVisitDate)) {
                            uvCt = 1L;
                            lastVisitDateState.update(curVisitDate);
                        }

                        Long svCt = StringUtils.isEmpty(pageJsonObj.getString("last_page_id")) ? 1L : 0L;

                        TrafficPageViewBean viewBean = new TrafficPageViewBean(
                                "",
                                "",
                                "",
                                commonJsonObj.getString("sc"),
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                uvCt,
                                svCt,
                                1L,
                                pageJsonObj.getLong("during_time"),
                                ts
                        );
                        out.collect(viewBean);
                    }
                }
        );
        // beanDS.print();
        /**
         * TrafficPageViewBean(stt=, edt=, cur_date=,
         *                      vc=v2.1.134, ch=Appstore, ar=5, isNew=0,
         *                      uvCt=0, svCt=0, pvCt=1, durSum=16877, ts=1731640512722)
         */

        //TODO 4.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TrafficPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewBean bean, long recordTimestamp) {
                                        return bean.getTs();
                                    }
                                }
                        )
        );
        // withWatermarkDS.print();

        //TODO 5.按照统计的维度进行分组
        KeyedStream<TrafficPageViewBean, Tuple5<String, String, String, String,String>> dimKeyedDS = withWatermarkDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple5<String, String, String, String,String>>() {
            @Override
            public Tuple5<String, String, String, String,String> getKey(TrafficPageViewBean bean) throws Exception {
                return Tuple5.of(
                        bean.getSc(),
                        bean.getVc(),
                        bean.getCh(),
                        bean.getAr(),
                        bean.getIsNew()
                );
            }
        });
        // dimKeyedDS.print();

        //TODO 6.开窗
        WindowedStream<TrafficPageViewBean, Tuple5<String, String, String, String,String>, TimeWindow> windowDS
                = dimKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 7.聚合
        /**
         * 关于FlinkAPI的处理函数
         *      增量聚合
         *          优势：数据来一条处理一次，不会缓存所有数据，节省空间   劣势：获取不到窗口详细信息
         *          reduce
         *              窗口中元素类型和向下游传递的类型一致
         *          aggregate
         *              窗口中元素类型、累加器类型以及向下游传递的数据类型可以不一致
         *      全量聚合
         *          优势：可以获取到窗口详细信息     劣势：会缓存所有数据，费空间
         *          apply
         *              参数接收的是窗口对象
         *          process
         *              参数接收的是上下文对象，除了可以获取窗口对象之外，还可以获取水位线相关信息   更底层
         *      增量 + 全量
         */
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                },
                /*new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {

                    }
                }*/
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple5<String, String, String, String,String>, TimeWindow>() {
                    @Override
                    public void process(Tuple5<String, String, String, String,String> stringStringStringStringTuple4, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple5<String, String, String, String,String>, TimeWindow>.Context context, Iterable<TrafficPageViewBean> elements, Collector<TrafficPageViewBean> out) throws Exception {
                        TimeWindow window = context.window();
                        TrafficPageViewBean viewBean = elements.iterator().next();
                        viewBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        viewBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        viewBean.setCur_date(DateFormatUtil.tsToDate(window.getStart()));
                        out.collect(viewBean);
                    }
                }
        );

        reduceDS.print();
        /**
         * TrafficPageViewBean(stt=2024-11-15 11:13:50,
         * edt=2024-11-15 11:14:00, cur_date=2024-11-15, vc=v2.1.132, ch=oppo, ar=23, isNew=0, uvCt=0,
         * svCt=0, pvCt=1, durSum=811733, ts=1731640432555)
         */
        //TODO 8.将聚合的结果写到Doris
        // reduceDS.print();
        reduceDS
                .map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));
    }
}
